package com.xbongbong.saas.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.xbongbong.paas.domain.entity.ext.PaasFormDataEntityExt;
import com.xbongbong.paas.elasticsearch.pojo.XbbAggregatedPage;
import com.xbongbong.paas.elasticsearch.util.EsHelper;
import com.xbongbong.paas.enums.DelEnum;
import com.xbongbong.paas.enums.FlowStatusEnum;
import com.xbongbong.paas.enums.IndexTypeEnum;
import com.xbongbong.paas.toolbox.exception.XbbException;
import com.xbongbong.paas.toolbox.util.BeanUtil;
import com.xbongbong.paas.toolbox.util.CommentUtil;
import com.xbongbong.paas.toolbox.util.DateTimeUtil;
import com.xbongbong.paas.toolbox.util.DateUtil;
import com.xbongbong.paas.toolbox.wrap.BaseVO;
import com.xbongbong.paas.toolbox.wrap.UserVO;
import com.xbongbong.pro.domain.entity.ScoreScriptLogEntity;
import com.xbongbong.pro.enums.ErrorCodeEnum;
import com.xbongbong.pro.enums.SaasMarkEnum;
import com.xbongbong.pro.enums.errorcodes.ScoreErrorCodeEnum;
import com.xbongbong.pro.enums.errorcodes.SystemErrorCodeEnum;
import com.xbongbong.pro.rabbitmq.producer.ScoreProducer;
import com.xbongbong.pro.score.pojo.dto.ScoreRabbitMqDTO;
import com.xbongbong.pro.score.pojo.dto.ScoreScripTtriggerDTO;
import com.xbongbong.pro.score.pojo.dto.ScoreScriptLogDTO;
import com.xbongbong.saas.constant.BasicConstant;
import com.xbongbong.saas.constant.StringConstant;
import com.xbongbong.saas.enums.ScoreScriptOperationTypeEnum;
import com.xbongbong.saas.enums.XbbRefTypeEnum;
import com.xbongbong.saas.model.IndexTypeModel;
import com.xbongbong.saas.model.ScoreScriptLogModel;
import com.xbongbong.saas.service.ScoreScriptService;
import com.xbongbong.sys.help.SysEnvHelp;
import org.apache.commons.collections4.CollectionUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
import static org.elasticsearch.index.query.QueryBuilders.rangeQuery;
import static org.elasticsearch.index.query.QueryBuilders.termQuery;
import static org.elasticsearch.index.query.QueryBuilders.termsQuery;


/**
 * @author chunyang.qin
 * @version v1.0
 * @date 2020/11/4 10:35
 * @since v1.0
 */
@Service("scoreScriptService")
public class ScoreScriptServiceImpl implements ScoreScriptService {
    private static final Logger LOG = LoggerFactory.getLogger(ScoreScriptServiceImpl.class);
    private static final Integer YEAR = 366 * 86400;
    private static final Integer MAX_COUNT = 50000;

    @Resource
    private ScoreScriptLogModel scoreScriptLogModel;
    @Resource
    private EsHelper esHelper;
    @Resource
    private ScoreProducer scoreProducer;
    @Resource
    private IndexTypeModel indexTypeModel;
    @Resource
    private SysEnvHelp sysEnvHelp;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public BaseVO saveScoreScript(ScoreScriptLogDTO scoreScriptLogDTO) throws XbbException {
        String corpid = scoreScriptLogDTO.getCorpid();
        Long formId = scoreScriptLogDTO.getFormId();
        Long startTime = scoreScriptLogDTO.getStartTime();
        Long endTime = scoreScriptLogDTO.getEndTime();
        Integer businessType = scoreScriptLogDTO.getBusinessType();
        UserVO loginUser = scoreScriptLogDTO.getLoginUser();
        if (Objects.isNull(loginUser)) {
            //没有登录 或者 没有走网关
            throw new XbbException(SystemErrorCodeEnum.API_ERROR_100001, "获取用户信息失败");
        }
        //数据总量限制
        verificationScoreScript(corpid, businessType, formId, startTime, endTime);
        //数据拼装
        ScoreScriptLogEntity scoreScriptLogEntity = new ScoreScriptLogEntity();
        BeanUtil.copyProperties(scoreScriptLogDTO, scoreScriptLogEntity);
        scoreScriptLogEntity.setCreatorId(loginUser.getUserId());
        //触发日期，用户今天点击的重新计算，在明天的时候触发
        Long triggerDate = Long.valueOf(DateUtil.getTodayInt() + 86400);
        //根据触发日期，查询一遍，判断是更新还是插入
        ScoreScriptLogEntity oldScoreScriptLog = scoreScriptLogModel.getByTriggerDateAndFormId(triggerDate, formId, corpid);
        if (Objects.nonNull(oldScoreScriptLog)) {
            //覆盖更新
            oldScoreScriptLog.setStartTime(scoreScriptLogDTO.getStartTime());
            oldScoreScriptLog.setEndTime(scoreScriptLogDTO.getEndTime());
            scoreScriptLogModel.update(oldScoreScriptLog);
        } else {
            //新增
            scoreScriptLogEntity.setTriggerDate(triggerDate);
            scoreScriptLogModel.insert(scoreScriptLogEntity);
        }
        return new BaseVO();
    }

    @Override
    public BaseVO triggerScoreScript(ScoreScripTtriggerDTO scoreScriptLogDTO) throws XbbException {
        BaseVO baseVO = new BaseVO();
        LOG.info("=============================== 触发评分脚本 ============================================" + DateTimeUtil.getDate());
        Integer operationCode = scoreScriptLogDTO.getOperationCode();
        String key = scoreScriptLogDTO.getKey();
        if (!Objects.equals(key, BasicConstant.TASK_SECRET_KEY)) {
            LOG.error("评分重新计算脚本触发校验 密码错误，停止定时任务！");
            throw new XbbException(SystemErrorCodeEnum.API_ERROR_100066);
        }
        ScoreScriptOperationTypeEnum scoreScriptLogTypeEnum = ScoreScriptOperationTypeEnum.getByCode(operationCode);
        if (Objects.equals(scoreScriptLogTypeEnum, ScoreScriptOperationTypeEnum.UNKNOWN)) {
            LOG.error("operationCode只能为0和1,0:正常触发  1:自定义触发");
            throw new XbbException(SystemErrorCodeEnum.API_ERROR_100066);
        }
        long now = DateUtil.getTodayInt();
        List<String> corpidList = scoreScriptLogDTO.getCorpidList();
        Integer businessType = scoreScriptLogDTO.getBusinessType();
        Long formId = scoreScriptLogDTO.getFormId();
        Long triggerDate = scoreScriptLogDTO.getTriggerDate();

        Map<String, Object> param = new HashMap<>();
        List<ScoreScriptLogEntity> scoreScriptLogList;
        param.put("del", 0);
        param.put("triggerDate", now);
        CommentUtil.addToMap(param, "corpidList", corpidList);
        CommentUtil.addToMap(param, "businessType", businessType);
        CommentUtil.addToMap(param, "formId", formId);
        switch (scoreScriptLogTypeEnum) {
            case NORMAL:
                //正常执行
                scoreScriptLogList = scoreScriptLogModel.findEntitys(param);
                break;
            case REPAIR:
                //重新触发
                //覆盖原来的触发日期
                CommentUtil.addToMap(param, "triggerDate", triggerDate);
                scoreScriptLogList = scoreScriptLogModel.findEntitys(param);
                break;
            default:
                throw new XbbException(SystemErrorCodeEnum.API_ERROR_100002, "operationCode 不能为空");
        }
        if (CollectionUtils.isEmpty(scoreScriptLogList)) {
            LOG.info("=============================== 触发日期为{}的评分脚本数为0,结束执行 ============================================" + DateTimeUtil.getDate());
            return baseVO;
        }
        boolean isGrayEnvironmentEnable = sysEnvHelp.isGrayEnvironmentEnable();
        for (ScoreScriptLogEntity scoreScriptLogEntity : scoreScriptLogList) {

            if (sysEnvHelp.isGrayEnvironment()) {
                if (!isGrayEnvironmentEnable) {
                    continue;
                }
                if (!sysEnvHelp.checkIsCrayCompany(scoreScriptLogEntity.getCorpid())) {
                    continue;
                }
            } else {
                if (isGrayEnvironmentEnable) {
                    if (sysEnvHelp.checkIsCrayCompany(scoreScriptLogEntity.getCorpid())) {
                        continue;
                    }
                }
            }

            try {
                handleSingleScoreScript(scoreScriptLogEntity);
                scoreScriptLogModel.deleteByKey(scoreScriptLogEntity.getId(),scoreScriptLogEntity.getCorpid());
            } catch (Exception e) {
                LOG.error("triggerScoreScript发生异常,原因{}", e.getMessage());
                continue;
            }
        }
        LOG.info("=============================== 评分生产者消息发送完毕,脚本已全部触发!============================================" + DateTimeUtil.getDate());
        return baseVO;
    }

    /**
     *
     * @param scoreScriptLogEntity
     * @return
     * @throws XbbException
     */
    @Override
    public BaseVO handleSingleScoreScript(ScoreScriptLogEntity scoreScriptLogEntity) throws XbbException {
        BaseVO baseVO = new BaseVO();

        String corpid = scoreScriptLogEntity.getCorpid();
        Long formId = scoreScriptLogEntity.getFormId();
        Long startTime = scoreScriptLogEntity.getStartTime();
        Long endTime = scoreScriptLogEntity.getEndTime();
        Integer businessType = scoreScriptLogEntity.getBusinessType();
        XbbRefTypeEnum xbRefTypeEnum = XbbRefTypeEnum.getByCode(businessType);
        IndexTypeEnum indexTypeEnum;
        switch (xbRefTypeEnum) {
            case CUSTOMER_MANAGEMENT:
                indexTypeEnum = indexTypeModel.getCustomerIndex(corpid, businessType, SaasMarkEnum.SAAS.getCode());
                break;
            case CLUE:
                indexTypeEnum = IndexTypeEnum.IDX_SAAS_CLUE;
                break;
            default:
                throw new XbbException(ScoreErrorCodeEnum.API_ERROR_1200001);
        }
        //数据总数
        Integer dataCount = getDataCount(indexTypeEnum, corpid, formId, startTime, endTime);
        if (dataCount == 0) {
            LOG.info("数据总量为0，脚本执行结束");
            return baseVO;
        }
        //超过1万条时，分页处理
        int pageSize = 10000;
        int page = dataCount / pageSize;
        if (dataCount % pageSize != 0) {
            page++;
        }
        for (int pageIndex = 0; pageIndex <= page; pageIndex++) {
            //i的0对应page的1
            handleProducerByPage(corpid, formId, startTime, endTime, ++pageIndex, pageSize, indexTypeEnum, businessType);
        }

        return baseVO;
    }

    /**
     * 分页处理，生产者发送消息
     */
    private void handleProducerByPage(String corpid, Long formId, Long startTime, Long endTime, int page, int pageSize, IndexTypeEnum indexTypeEnum, Integer businessType) throws XbbException {
        //查询出需要"重新计算"的评分数据主键集合
        BoolQueryBuilder boolQueryBuilder = boolQuery();
        packageBoolQuery(boolQueryBuilder, corpid, formId, startTime, endTime);
        //页数是从 1 ，开始的
        XbbAggregatedPage<PaasFormDataEntityExt> aggregatedPage = esHelper.findByPage(indexTypeEnum, boolQueryBuilder, PaasFormDataEntityExt.class, Arrays.asList("dataId"), page, pageSize, null);
        if (Objects.nonNull(aggregatedPage)) {
            List<PaasFormDataEntityExt> dataEntityExtList = aggregatedPage.getContent();
            if (CollectionUtils.isNotEmpty(dataEntityExtList)) {
                //每页1000条，同时每次消息发送也是塞入1000个dataId
                Integer msgPagerSize = 1000;
                //本次查询的总数据量
                int dataCount = dataEntityExtList.size();
                //初始页码为 0
                Integer msgPage = 0;
                List<PaasFormDataEntityExt> subList;
                while (dataCount > 0) {
                    if (dataCount > msgPagerSize) {
                        //至少2页
                        try {
                            subList = dataEntityExtList.subList(page * msgPagerSize, (page + 1) * msgPagerSize);
                            //生产者发送消息
                            scoreProducerSend(businessType, corpid, subList);
                        } catch (Exception e) {
                            LOG.error("handleSingleScoreScript发生异常,原因:", e);
                        } finally {
                            page++;
                            dataCount = dataCount - msgPagerSize;
                        }
                        // 如果需要在此处增加代码，需要将finally代码加入到catch中并增加continue
                    } else {
                        //1页之内
                        try {
                            scoreProducerSend(businessType, corpid, dataEntityExtList);
                        } catch (Exception e) {
                            LOG.error("handleSingleScoreScript发生异常,原因:", e);
                        } finally {
                            dataCount = 0;
                        }
                        // 如果需要在此处增加代码，需要将finally代码加入到catch中并增加continue
                    }
                }
            }


        }

    }

    /**
     * 生产者发送消息
     */
    private void scoreProducerSend(Integer businessType, String corpid, List<PaasFormDataEntityExt> dataEntityExtList) {
        if (CollectionUtils.isNotEmpty(dataEntityExtList)) {
            List<Long> dataIdList = dataEntityExtList.stream().map(item -> item.getDataId()).collect(Collectors.toList());
            ScoreRabbitMqDTO msg = new ScoreRabbitMqDTO(businessType, corpid, dataIdList);
            scoreProducer.sendMessage(msg);
        }
    }


    /**
     * 获取数据总量
     *
     * @return
     */
    private Integer getDataCount(IndexTypeEnum indexTypeEnum, String corpid, Long formId, Long startTime, Long endTime) throws XbbException {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQueryBuilder = boolQuery();
        //查询条件封装
        packageBoolQuery(boolQueryBuilder, corpid, formId, startTime, endTime);
        sourceBuilder.query(boolQueryBuilder);
        //数量聚合
        ValueCountAggregationBuilder valueCountAggregationBuilder = AggregationBuilders.count("aggs01").field(StringConstant.DATA_ID);
        JSONObject resultJsonObject = esHelper.elasticAggs4Buckets2Object(indexTypeEnum, boolQueryBuilder, valueCountAggregationBuilder);
        Long total = resultJsonObject.getLong("total");
        return total == null ? 0 : total.intValue();
    }


    /**
     * 封装es查询条件
     */
    private void packageBoolQuery(BoolQueryBuilder boolQueryBuilder, String corpid, Long formId, Long startTime, Long endTime) {
        boolQueryBuilder.filter(termQuery(StringConstant.CORPID + StringConstant.POINT + StringConstant.KEY_WORD, corpid));
        //时间筛选
        boolQueryBuilder.filter(rangeQuery("updateTime").gte(startTime).lte(endTime));
        //表单筛选
        boolQueryBuilder.filter(termQuery(StringConstant.FORM_ID, formId));
        //审核状态
        boolQueryBuilder.filter(termsQuery("flowStatus", Arrays.asList(FlowStatusEnum.NOT_NEED_APPROVAL.getType(), FlowStatusEnum.PASS.getType())));
        boolQueryBuilder.filter(termQuery(StringConstant.DEL, DelEnum.NORMAL.getDel()));
    }


    /**
     * 数据总量限制
     *
     * @param corpid
     * @param formId
     * @param startTime
     * @param endTime
     */
    private void verificationScoreScript(String corpid, Integer businessType, Long formId, Long startTime, Long endTime) throws XbbException {
        //时间范围最多选择1年
        if ((endTime - startTime) > YEAR) {
            throw new XbbException(ScoreErrorCodeEnum.API_ERROR_1200017, ScoreErrorCodeEnum.API_ERROR_1200017.getMsg());
        }
        XbbRefTypeEnum xbRefTypeEnum = XbbRefTypeEnum.getByCode(businessType);
        IndexTypeEnum indexTypeEnum;
        switch (xbRefTypeEnum) {
            case CUSTOMER_MANAGEMENT:
                indexTypeEnum = indexTypeModel.getCustomerIndex(corpid, XbbRefTypeEnum.CUSTOMER_MANAGEMENT.getCode(), SaasMarkEnum.SAAS.getCode());
                break;
            case CLUE:
                indexTypeEnum = IndexTypeEnum.IDX_SAAS_CLUE;
                break;
            default:
                throw new XbbException(ScoreErrorCodeEnum.API_ERROR_1200001, ScoreErrorCodeEnum.API_ERROR_1200001.getMsg());
        }
        //数据总量限制
        Integer dataCount = getDataCount(indexTypeEnum, corpid, formId, startTime, endTime);
        if (dataCount > MAX_COUNT) {
            throw new XbbException(ScoreErrorCodeEnum.API_ERROR_1200018, ScoreErrorCodeEnum.API_ERROR_1200018.getMsg());
        }
    }

}
